home *** CD-ROM | disk | FTP | other *** search
/ Personal Computer World 2008 February / PCWFEB08.iso / Software / Freeware / Miro 1.0 / Miro_Installer.exe / Miro_Downloader.exe / test / httpdownloadertest.pyc (.txt) < prev    next >
Encoding:
Python Compiled Bytecode  |  2007-11-12  |  6.7 KB  |  178 lines

  1. # Source Generated with Decompyle++
  2. # File: in.pyc (Python 2.5)
  3.  
  4. import os
  5. import tempfile
  6. import util
  7. import download_utils
  8. import httpclient
  9. from test import schedulertest
  10. from dl_daemon import download
  11.  
  12. def testingNextFreeFilename(filename):
  13.     return tempfile.mktemp()
  14.  
  15.  
  16. class TestingDownloader(download.HTTPDownloader):
  17.     UPDATE_CLIENT_INTERVAL = 0
  18.     
  19.     def __init__(self, test, *args, **kwargs):
  20.         self.test = test
  21.         if 'restore' in kwargs:
  22.             
  23.             kwargs['restore']['statusCallback'] = lambda : 0
  24.             kwargs['restore']['lastStatus'] = None
  25.         else:
  26.             
  27.             self.statusCallback = lambda : 0
  28.             self.lastStatus = None
  29.         download.HTTPDownloader.__init__(self, *args, **kwargs)
  30.  
  31.     
  32.     def updateClient(self):
  33.         self.lastStatus = self.getStatus()
  34.         self.test.addIdle(self.statusCallback, 'status callback')
  35.  
  36.  
  37.  
  38. class HTTPDownloaderTest(schedulertest.EventLoopTest):
  39.     
  40.     def setUp(self):
  41.         super(HTTPDownloaderTest, self).setUp()
  42.         download.chatter = False
  43.         download.nextFreeFilename = testingNextFreeFilename
  44.         download._downloads = { }
  45.         allConnections = []
  46.         for conns in httpclient.HTTPClient.connectionPool.connections.values():
  47.             allConnections.extend(conns['active'])
  48.             allConnections.extend(conns['free'])
  49.         
  50.         for c in allConnections:
  51.             c.closeConnection()
  52.         
  53.         httpclient.HTTPClient.connectionPool = httpclient.HTTPConnectionPool()
  54.  
  55.     
  56.     def tearDown(self):
  57.         download.nextFreeFilename = download_utils.nextFreeFilename
  58.         download.chatter = True
  59.         super(HTTPDownloaderTest, self).tearDown()
  60.  
  61.     
  62.     def stopOnFinished(self):
  63.         if self.downloader.state == 'finished':
  64.             self.stopEventLoop(False)
  65.         
  66.  
  67.     
  68.     def getDownloadedData(self):
  69.         return open(self.downloader.filename, 'rb').read()
  70.  
  71.     
  72.     def countConnections(self):
  73.         total = 0
  74.         pool = httpclient.HTTPClient.connectionPool
  75.         for conns in pool.connections.values():
  76.             total += len(conns['active'])
  77.         
  78.         return total
  79.  
  80.     
  81.     def testDownload(self):
  82.         url = u'http://participatoryculture.org/democracytest/normalpage.txt'
  83.         self.downloader = TestingDownloader(self, url, 'ID1')
  84.         self.downloader.statusCallback = self.stopOnFinished
  85.         self.runEventLoop()
  86.         self.assertEquals(self.getDownloadedData(), 'I AM A NORMAL PAGE\n')
  87.  
  88.     
  89.     def testStop(self):
  90.         url = u'http://www.getdemocracy.com/images/linux-screen.jpg'
  91.         self.downloader = TestingDownloader(self, url, 'ID1')
  92.         
  93.         def stopOnData():
  94.             if self.downloader.state == 'downloading' and self.downloader.currentSize > 0:
  95.                 self.downloader.stop(False)
  96.                 self.stopEventLoop(False)
  97.             
  98.  
  99.         self.downloader.statusCallback = stopOnData
  100.         self.runEventLoop()
  101.         self.assertEquals(self.downloader.state, 'stopped')
  102.         self.assertEquals(self.downloader.currentSize, 0)
  103.         self.assert_(not os.path.exists(self.downloader.filename))
  104.         self.assertEquals(self.countConnections(), 0)
  105.         
  106.         def restart():
  107.             self.downloader.start()
  108.  
  109.         self.addTimeout(0.5, restart, 'restarter')
  110.         self.downloader.statusCallback = self.stopOnFinished
  111.         self.runEventLoop()
  112.         self.assertEquals(self.downloader.currentSize, 45572)
  113.         self.assertEquals(self.downloader.totalSize, 45572)
  114.  
  115.     
  116.     def testPause(self):
  117.         url = u'http://www.getdemocracy.com/images/linux-screen.jpg'
  118.         self.downloader = TestingDownloader(self, url, 'ID1')
  119.         
  120.         def pauseOnData():
  121.             if self.downloader.state == 'downloading' and self.downloader.currentSize > 0:
  122.                 self.downloader.pause()
  123.                 self.stopEventLoop(False)
  124.             
  125.  
  126.         self.downloader.statusCallback = pauseOnData
  127.         self.runEventLoop()
  128.         self.assertEquals(self.downloader.state, 'paused')
  129.         self.assert_(self.downloader.currentSize > 0)
  130.         self.assert_(os.path.exists(self.downloader.filename))
  131.         self.assertEquals(self.countConnections(), 0)
  132.         
  133.         def restart():
  134.             self.downloader.start()
  135.  
  136.         self.addTimeout(0.5, restart, 'restarter')
  137.         self.downloader.statusCallback = self.stopOnFinished
  138.         self.runEventLoop()
  139.         self.assertEquals(self.downloader.currentSize, 45572)
  140.         self.assertEquals(self.downloader.totalSize, 45572)
  141.  
  142.     
  143.     def testRestore(self):
  144.         url = u'http://www.getdemocracy.com/images/linux-screen.jpg'
  145.         self.downloader = TestingDownloader(self, url, 'ID1')
  146.         
  147.         def pauseInMiddle():
  148.             if self.downloader.state == 'downloading' and self.downloader.currentSize > 1000:
  149.                 self.downloader.pause()
  150.                 self.stopEventLoop(False)
  151.             
  152.  
  153.         self.downloader.statusCallback = pauseInMiddle
  154.         self.runEventLoop()
  155.         self.assertEquals(self.downloader.state, 'paused')
  156.         (None,)(self.assert_ if self.downloader.currentSize < self.downloader.currentSize else self.downloader.currentSize < 2000)
  157.         restore = self.downloader.lastStatus.copy()
  158.         restore['state'] = 'downloading'
  159.         download._downloads = { }
  160.         self.downloader2 = TestingDownloader(self, restore = restore)
  161.         restoreSize = restore['currentSize']
  162.         self.restarted = False
  163.         
  164.         def statusCallback():
  165.             if self.downloader2.currentSize < restoreSize:
  166.                 print '%d < %d' % (self.downloader2.currentSize, restoreSize)
  167.                 self.restarted = True
  168.                 self.stopEventLoop(False)
  169.             elif self.downloader2.state == 'finished':
  170.                 self.stopEventLoop(False)
  171.             
  172.  
  173.         self.downloader2.statusCallback = statusCallback
  174.         self.runEventLoop()
  175.         self.assert_(not (self.restarted))
  176.  
  177.  
  178.